package com.lagou.work4;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;

public class WaterMarkDemo {
    public static void main(String[] args) throws Exception {

        //设置flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.getConfig().setAutoWatermarkInterval(1000L);
        env.setParallelism(1);
        //  1、获取数据源
        DataStreamSource<String> data = env.socketTextStream("server1",9099);
        //  2、转化
        SingleOutputStreamOperator<Tuple2<String,Long>> map = data.map(
                new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String s) throws Exception {
                        String[] s1 = s.split(",");
                        return new Tuple2<>(s1[0],Long.valueOf(s1[1]));
                    }
                }
        );
        //  3、声明水印（watermark）
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-mm HH-mm-ss.SSS");
        SingleOutputStreamOperator<Tuple2<String, Long>> watermarks = map.assignTimestampsAndWatermarks(
                new WatermarkStrategy<Tuple2<String, Long>>() {

                    @Override
                    public WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                        return new WatermarkGenerator<Tuple2<String, Long>>() {
                            private long maxTimeStamp = Long.MIN_VALUE;
                            @Override
                            public void onEvent(Tuple2<String, Long> stringLongTuple2, long l, WatermarkOutput watermarkOutput) {
                                maxTimeStamp = Math.max(maxTimeStamp,stringLongTuple2.f1);
                                System.out.println("maxTimeStamp:" + maxTimeStamp + "...format:" + sdf.format(maxTimeStamp));
                            }

                            @Override
                            public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
                                System.out.println("……onPeriodicEmit……");
                                long maxOutOfOrderness = 3000l;
                                watermarkOutput.emitWatermark((new Watermark(maxTimeStamp - maxOutOfOrderness)));

                            }
                        };
                    }
                }
        );
        //  4、分组聚合，调用window的操作
        KeyedStream<Tuple2<String, Long>, String> keyedStream = watermarks.keyBy(value -> value.f0);
        WindowedStream<Tuple2<String, Long>, String, TimeWindow> windowed = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(4)));
        SingleOutputStreamOperator<String> outputStreamOperator = windowed.apply(
                new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<Tuple2<String, Long>> iterable, Collector<String> collector) throws Exception {
                        String key = s;
                        Iterator<Tuple2<String, Long>> tuple2Iterator = iterable.iterator();
                        ArrayList<Long> list = new ArrayList<>();
                        while (tuple2Iterator.hasNext()){
                            Tuple2<String, Long> next = tuple2Iterator.next();
                            list.add(next.f1);
                        }
                        //排序
                        Collections.sort(list);
                        //将结果输出 key是什么 数值是多少 窗口大小
                        String result = "key:" + key + "," + "list.first:" + sdf.format(list.get(0)) + "," + "list.last" + sdf.format(list.get(list.size() - 1)) + "," + "window.start" + sdf.format(timeWindow.getStart()) + ","  + "window.end" + sdf.format(timeWindow.getEnd());
                        collector.collect(result);

                    }
                }
        );
        //  5、保存处理结果
        outputStreamOperator.print();
        env.execute();
    }
}
